/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <cstdio>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"
#include <limits>

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);

RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

// Helper functions for SqlEngine::select
static int 
_findKeyLowerBound(const vector<SelCond>& cond);

static int
_findKeyUpperBound(const vector<SelCond>& cond);

static bool
_queryDependsOnNonIndexedAttr(int attr, const vector<SelCond>& cond);

static RC
_indexSelect(int attr, const vector<SelCond>& cond, RecordFile& rf, const string& table);

static bool
_tupleMatches(const vector<SelCond>& cond, RecordFile& rf, RecordId rid, int key);

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  RecordFile rf;    // RecordFile containing the table
  RecordId   rid;   // record cursor for table scanning

  RC     rc;
  int    key;     
  string value;
  int    count;
  int    diff;

  // open the table file
  if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
    fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
    return rc;
  }

  // try to use the index
  if ((rc = _indexSelect(attr, cond, rf, table)) != RC_USE_DIRECT_SCAN) {
	  goto exit_select;
  } 

  // scan the table file from the beginning
  rid.pid = rid.sid = 0;
  count = 0;
  while (rid < rf.endRid()) {
    // read the tuple
    if ((rc = rf.read(rid, key, value)) < 0) {
      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
      goto exit_select;
    }

    // check the conditions on the tuple
    for (unsigned i = 0; i < cond.size(); i++) {
      // compute the difference between the tuple value and the condition value
      switch (cond[i].attr) {
      case 1:
	diff = key - atoi(cond[i].value);
	break;
      case 2:
	diff = strcmp(value.c_str(), cond[i].value);
	break;
      }

      // skip the tuple if any condition is not met
      switch (cond[i].comp) {
      case SelCond::EQ:
	if (diff != 0) goto next_tuple;
	break;
      case SelCond::NE:
	if (diff == 0) goto next_tuple;
	break;
      case SelCond::GT:
	if (diff <= 0) goto next_tuple;
	break;
      case SelCond::LT:
	if (diff >= 0) goto next_tuple;
	break;
      case SelCond::GE:
	if (diff < 0) goto next_tuple;
	break;
      case SelCond::LE:
	if (diff > 0) goto next_tuple;
	break;
      }
    }

    // the condition is met for the tuple. 
    // increase matching tuple counter
    count++;

    // print the tuple 
    switch (attr) {
    case 1:  // SELECT key
      fprintf(stdout, "%d\n", key);
      break;
    case 2:  // SELECT value
      fprintf(stdout, "%s\n", value.c_str());
      break;
    case 3:  // SELECT *
      fprintf(stdout, "%d '%s'\n", key, value.c_str());
      break;
    }

    // move to the next tuple
    next_tuple:
    ++rid;
  }

  // print matching tuple count if "select count(*)"
  if (attr == 4) {
    fprintf(stdout, "%d\n", count);
  }
  rc = 0;

  // close the table file and return
  exit_select:
  rf.close();
  return rc;
}

RC _indexSelect(int attr, const vector<SelCond>& cond, RecordFile& rf, const string& table) {

	RC rc;
	BTreeIndex index;
	int keyLowerBound = _findKeyLowerBound(cond);
	int keyUpperBound = _findKeyUpperBound(cond);
	bool hasLowerBound = keyLowerBound > numeric_limits<int>::min();
	bool hasUpperBound = keyUpperBound < numeric_limits<int>::max();

	if (keyLowerBound == keyUpperBound) {

		//There exists an equality condition on the key.
		int key;
		IndexCursor cursor;
		RecordId tupleRid;
		string value;

		if (index.open(table + ".idx", 'r')) {
			//No index exists
			return RC_USE_DIRECT_SCAN;
		}
		
		rc = index.locate(keyLowerBound, cursor);
		if (rc == RC_END_OF_TREE) {
			index.close();
			return 0; //no matching tuple
		}
		else if (rc) {
			index.close();
			return rc; //error
		}

		//read the index leaf-node
		if ((rc = index.readForward(cursor, key, tupleRid)) != 0) {
			index.close();
			return rc;
		}

		if (_tupleMatches(cond, rf, tupleRid, key)) {
			switch(attr) {
				case 1:
					fprintf(stdout, "%d\n", key);
					break;
				case 2:
					rf.read(tupleRid, key, value);
					fprintf(stdout, "%s\n", value.c_str());
					break;
				case 3:
					rf.read(tupleRid, key, value);
					fprintf(stdout, "%d '%s'\n", key, value.c_str());				
					break;
				case 4:
					fprintf(stdout, "%d\n", 1);
					break;
				default:
					index.close();
					return RC_INVALID_ATTRIBUTE;
			}
		}

		index.close();
		return 0;
	}
	else if (hasLowerBound || hasUpperBound) {
		//There exists an inequality condition on the key.
		int count = 0;
		string value;
		IndexCursor cursor;
		int key;
		RecordId rid;

		if (index.open(table + ".idx", 'r')) {
			//No index exists
			return RC_USE_DIRECT_SCAN;
		}

		//Locate the lower bound
		rc = index.locate(keyLowerBound, cursor);
		if (rc == RC_END_OF_TREE) {
			index.close();
			return 0;  //No tuples match
		}
		else if (rc) {
			index.close();
			return rc; //error
		}
		
		while (true) {

			//Read an index entry
			rc = index.readForward(cursor, key, rid);
			if (rc == RC_END_OF_TREE
			    || (rc == 0 && key > keyUpperBound))
				break; //done reading tuples
			else if (rc) {
				index.close();
				return rc; //error
			}

			if (_tupleMatches(cond, rf, rid, key)) {
				count++;
				switch(attr) {
					case 1:
						fprintf(stdout, "%d\n", key);
						break;
					case 2:
						rf.read(rid, key, value);
						fprintf(stdout, "%s\n", value.c_str());
						break;
					case 3:
						rf.read(rid, key, value);
						fprintf(stdout, "%d '%s'\n", key, value.c_str());				
						break;
				}
			}
		}
		
		if (attr == 4) {
			fprintf(stdout, "%d\n", count);
		}

		index.close();
		return 0;
	}
	else if ( ! _queryDependsOnNonIndexedAttr(attr, cond) ) {
		int key = 0;
		IndexCursor cursor;

		if (index.open(table + ".idx", 'r')) {
			//No index exists
			return RC_USE_DIRECT_SCAN;
		}

		rc = index.locate(keyLowerBound, cursor);
		if (rc) {
			index.close();
			return rc; //error, or index is empty
		}

		RecordId rid;	
			
		if (attr == 4) {
			int count = 0;
			RC rc;
			while ((rc = index.readForward(cursor, key, rid)) == 0) {
				count++;
			}
			fprintf(stdout, "%d\n", count);
		}
		else { //attr is 1
			while (index.readForward(cursor, key, rid) == 0) {
					if (_tupleMatches(cond, rf, rid, key)) {
						fprintf(stdout, "%d\n", key);
					}
				}
			}

		index.close();
		return 0;
	}
	else {
		//don't use the index.  do regular select
		return RC_USE_DIRECT_SCAN;
	}
}

/**
 *  Given the conditions cond, find the minimum value that
 *  a key must be in order to satisfy all conditions.  If
 *  there are no inequality conditions on the key, then 
 *  the minimmum value of a signed 32-bit integer is returned.
 *  @param cond: the conditions to check
 *  @returns: the minimum value of a key.
 */
int _findKeyLowerBound(const vector<SelCond>& cond) {

	int min = numeric_limits<int>::min();
	int currVal;
	
	for (unsigned i = 0; i < cond.size(); i++) {
		if (cond[i].attr == 1) {
			currVal = atoi(cond[i].value);
			if (cond[i].comp == SelCond::EQ && currVal > min) {
					min = currVal;
			}
			else if (cond[i].comp == SelCond::GT && currVal >= min) {
					min = currVal + 1;			
			}
			else if (cond[i].comp == SelCond::GE && currVal > min) {
					min = currVal;
			}
		}
	}
	
	return min;
}

/**
 *  Given the conditions cond, find the maximum value that
 *  a key can be in order to satisfy all conditions.
 *	@param cond: the conditions to check
 *  @returns: the maximum value of a key, or the maximum
 *		value of a signed 32-bit integer if there is no
 *      condition specifying a maximum key.
 */
int _findKeyUpperBound(const vector<SelCond>& cond) {

	int max = numeric_limits<int>::max();
	int currVal;
	
	for (unsigned i = 0; i < cond.size(); i++) {
		if (cond[i].attr == 1) {
			currVal = atoi(cond[i].value);
			if (cond[i].comp == SelCond::EQ && currVal < max) {
					max = currVal;
			}
			else if (cond[i].comp == SelCond::LT && currVal <= max) {
					max = currVal - 1;			
			}
			else if (cond[i].comp == SelCond::LE && currVal < max) {
					max = currVal;
			}
		}
	}

	return max;
}

/**
 *  Returns true if and only if the result of a query depends
 *  on non-indexed attributes.
 *  @param attr: attribute in the SELECT clause
 *  @param cond: list of conditions in the WHERE clause
 *  returns: true/false
 */
bool _queryDependsOnNonIndexedAttr(int attr, const vector<SelCond>& cond) {

	if (attr == 2 || attr == 3)
		return true;
	else {
		for (unsigned i = 0; i < cond.size(); i++) {
			if (cond[i].attr == 2)
				return true;
		}
	}
	return false;
}

/**
 *  Returns true if a tuple satisfies all conditions.  If this
 *  can be determined directly from the tuple's key, i.e without
 *  loading it from disk, then the tuple shall not be loaded from disk.
 *  @param cond: the list of conditions
 *	@param rf: the RecordFile containing the tuple
 *  @param rid: the record id of the tuple
 *  @param key: the tuple's key attribute
 *  @returns: true if the tuple satisfies all conditions.
 */
bool _tupleMatches(const vector<SelCond>& cond, RecordFile& rf, RecordId rid, int key) {

	//TODO: implement this function.  Don't do any unnecessary I/O
	for (unsigned i = 0; i < cond.size(); i++) {
		int diff;
		if (cond[i].attr == 1) {
			diff = key - atoi(cond[i].value);
		}
		else if (cond[i].attr == 2) {
			string tupleVal;
			int tupleKey;
			if (rf.read(rid, tupleKey, tupleVal) != 0) {
				fprintf(stderr, "Error: while reading a tuple from table\n");
				return false;
			}
			diff = strcmp(tupleVal.c_str(), cond[i].value);
		}
	
		switch(cond[i].comp) {
			case SelCond::EQ:
				if (diff != 0)
					return false;
				break;
			case SelCond::NE:
				if (diff == 0)
					return false;
				break;
			case SelCond::LT:
				if (diff >= 0)
					return false;
				break;
			case SelCond::GT:
				if (diff <= 0)
					return false;
				break;
			case SelCond::LE:
				if (diff > 0)
					return false;
				break;
			case SelCond::GE:
				if (diff < 0)
					return false;
				break;		
			}
	}
	
	return true;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
  /* your code here */
  
  RecordFile rf;
  BTreeIndex bti;
  ifstream loadFileStream;
  std::string tableFile = table + ".tbl";
  std::string indexFile;
  RC status;
  
  status = rf.open(tableFile, 'w'); 
  if (status) {
	fprintf(stderr, "Error: Failed to open record file \"%s\"\n", tableFile.c_str());
    return status; 
  }
  
  loadFileStream.open(loadfile.c_str(), ifstream::in);
  if (loadFileStream.fail()) {
	status = RC_FILE_OPEN_FAILED;
	fprintf(stderr, "Error: Failed to open file stream for load file \"%s\"\n", loadfile.c_str());
	goto exit;
  }
  
  if (index) {
	indexFile = table + ".idx";
	status = bti.open(indexFile, 'w');
	if (status) {
		fprintf(stderr, "Error: Failed to open record file \"%s\"\n", indexFile.c_str());
		goto exit;	
	}
  }  
  
  char line[10240];
  while (true) {
	loadFileStream.getline(line, 10240);
	if (loadFileStream.eof()) {
		//Done reading the file
		break;
	}
	
	if (loadFileStream.fail() || loadFileStream.bad()) {
		status = RC_FILE_READ_FAILED;
		fprintf(stderr, "Error while reading load file \"%s\"\n", loadfile.c_str());
		goto exit2;
	}
	
	string s_line (line);
	int key;
	string value;
	
	status = parseLoadLine(s_line, key, value);
	if (status !=0) {
		fprintf(stderr, "Error while parsing load file \"%s\"\n", loadfile.c_str());
		goto exit2;
		}
	
	RecordId rid;
	status = rf.append(key, value, rid);
	if (status != 0) {
		fprintf(stderr, "Error while writing to table \"%s\"\n", tableFile.c_str());
		goto exit2;
		}
	
	if (index) {
		status = bti.insert(key, rid);
		if (status != 0) {
			fprintf(stderr, "Error while writing to index \"%s\"\n", indexFile.c_str());
			goto exit2;			
		}
	}
	
  }
	
  
  exit2:
  loadFileStream.close();
  if (index)
	bti.close();
  exit:
  rf.close();
  return status;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
